const std = @import("std");
const Allocator = std.mem.Allocator;
const virtual_actor = @import("virtual_actor.zig");

/// 节点信息
pub const NodeInfo = struct {
    id: []const u8,
    address: []const u8,
    port: u16,
    status: NodeStatus,
    last_heartbeat: i64,
};

pub const NodeStatus = enum {
    active,
    inactive,
    failed,
};

/// 一致性哈希路由 - 用于分布式 Actor 定位
pub const ConsistentHashRouter = struct {
    allocator: Allocator,
    nodes: std.ArrayList(NodeInfo),
    hash_ring: std.AutoHashMap(u64, usize),
    virtual_nodes: usize,
    
    pub fn init(allocator: Allocator, virtual_nodes: usize) ConsistentHashRouter {
        return .{
            .allocator = allocator,
            .nodes = std.ArrayList(NodeInfo).init(allocator),
            .hash_ring = std.AutoHashMap(u64, usize).init(allocator),
            .virtual_nodes = virtual_nodes,
        };
    }
    
    pub fn deinit(self: *ConsistentHashRouter) void {
        self.nodes.deinit();
        self.hash_ring.deinit();
    }
    
    /// 添加节点
    pub fn addNode(self: *ConsistentHashRouter, node: NodeInfo) !void {
        const node_index = self.nodes.items.len;
        try self.nodes.append(node);
        
        // 为每个节点创建多个虚拟节点
        var i: usize = 0;
        while (i < self.virtual_nodes) : (i += 1) {
            const key = try std.fmt.allocPrint(
                self.allocator,
                "{s}:{d}",
                .{node.id, i}
            );
            defer self.allocator.free(key);
            
            const hash_value = std.hash.Wyhash.hash(0, key);
            try self.hash_ring.put(hash_value, node_index);
        }
    }
    
    /// 移除节点
    pub fn removeNode(self: *ConsistentHashRouter, node_id: []const u8) !void {
        // 找到节点索引
        var node_index: ?usize = null;
        for (self.nodes.items, 0..) |node, i| {
            if (std.mem.eql(u8, node.id, node_id)) {
                node_index = i;
                break;
            }
        }
        
        if (node_index == null) return error.NodeNotFound;
        
        // 从哈希环中移除虚拟节点
        var i: usize = 0;
        while (i < self.virtual_nodes) : (i += 1) {
            const key = try std.fmt.allocPrint(
                self.allocator,
                "{s}:{d}",
                .{node_id, i}
            );
            defer self.allocator.free(key);
            
            const hash_value = std.hash.Wyhash.hash(0, key);
            _ = self.hash_ring.remove(hash_value);
        }
        
        // 从节点列表中移除
        _ = self.nodes.swapRemove(node_index.?);
    }
    
    /// 获取 Actor 应该在哪个节点
    pub fn getNode(self: *ConsistentHashRouter, actor_id: []const u8) !NodeInfo {
        if (self.nodes.items.len == 0) return error.NoNodesAvailable;
        
        const hash_value = std.hash.Wyhash.hash(0, actor_id);
        
        // 在哈希环上查找最近的节点
        var min_distance: u64 = std.math.maxInt(u64);
        var selected_index: usize = 0;
        
        var iter = self.hash_ring.iterator();
        while (iter.next()) |entry| {
            const distance = if (entry.key_ptr.* >= hash_value)
                entry.key_ptr.* - hash_value
            else
                (std.math.maxInt(u64) - hash_value) + entry.key_ptr.*;
            
            if (distance < min_distance) {
                min_distance = distance;
                selected_index = entry.value_ptr.*;
            }
        }
        
        return self.nodes.items[selected_index];
    }
};

/// 分布式 Actor 代理 - 位置透明的远程调用
pub fn DistributedActorProxy(comptime ActorType: type) type {
    return struct {
        const Self = @This();
        
        allocator: Allocator,
        actor_id: []const u8,
        router: *ConsistentHashRouter,
        local_actor: ?*ActorType,
        
        pub fn init(
            allocator: Allocator,
            actor_id: []const u8,
            router: *ConsistentHashRouter,
        ) !Self {
            return .{
                .allocator = allocator,
                .actor_id = try allocator.dupe(u8, actor_id),
                .router = router,
                .local_actor = null,
            };
        }
        
        pub fn deinit(self: *Self) void {
            self.allocator.free(self.actor_id);
        }
        
        /// 调用 Actor 方法（位置透明）
        pub fn call(
            self: *Self,
            comptime method: []const u8,
            args: anytype,
        ) !void {
            // 获取 Actor 所在节点
            const node = try self.router.getNode(self.actor_id);
            
            if (self.isLocalNode(node)) {
                // 本地调用
                return self.callLocal(method, args);
            } else {
                // 远程调用
                return self.callRemote(node, method, args);
            }
        }
        
        fn isLocalNode(self: *Self, node: NodeInfo) bool {
            _ = self;
            _ = node;
            // TODO: 实现本地节点检测
            return true;
        }
        
        fn callLocal(
            self: *Self,
            comptime method: []const u8,
            args: anytype,
        ) !void {
            _ = self;
            _ = method;
            _ = args;
            // TODO: 实现本地方法调用
        }
        
        fn callRemote(
            self: *Self,
            node: NodeInfo,
            comptime method: []const u8,
            args: anytype,
        ) !void {
            _ = self;
            _ = node;
            _ = method;
            _ = args;
            // TODO: 实现远程 RPC 调用
        }
    };
}

/// 集群成员管理 - Gossip 协议
pub const ClusterMembership = struct {
    allocator: Allocator,
    local_node: NodeInfo,
    nodes: std.ArrayList(NodeInfo),
    mutex: std.Thread.Mutex,
    
    pub fn init(allocator: Allocator, local_node: NodeInfo) ClusterMembership {
        return .{
            .allocator = allocator,
            .local_node = local_node,
            .nodes = std.ArrayList(NodeInfo).init(allocator),
            .mutex = .{},
        };
    }
    
    pub fn deinit(self: *ClusterMembership) void {
        self.nodes.deinit();
    }
    
    /// 添加节点
    pub fn addNode(self: *ClusterMembership, node: NodeInfo) !void {
        self.mutex.lock();
        defer self.mutex.unlock();
        
        try self.nodes.append(node);
    }
    
    /// 移除节点
    pub fn removeNode(self: *ClusterMembership, node_id: []const u8) !void {
        self.mutex.lock();
        defer self.mutex.unlock();
        
        for (self.nodes.items, 0..) |node, i| {
            if (std.mem.eql(u8, node.id, node_id)) {
                _ = self.nodes.swapRemove(i);
                return;
            }
        }
    }
    
    /// 更新节点心跳
    pub fn updateHeartbeat(self: *ClusterMembership, node_id: []const u8) !void {
        self.mutex.lock();
        defer self.mutex.unlock();
        
        for (self.nodes.items) |*node| {
            if (std.mem.eql(u8, node.id, node_id)) {
                node.last_heartbeat = std.time.timestamp();
                node.status = .active;
                return;
            }
        }
    }
    
    /// 检测失败节点
    pub fn detectFailures(self: *ClusterMembership, timeout: i64) !std.ArrayList([]const u8) {
        self.mutex.lock();
        defer self.mutex.unlock();
        
        var failed = std.ArrayList([]const u8).init(self.allocator);
        const now = std.time.timestamp();
        
        for (self.nodes.items) |*node| {
            if (now - node.last_heartbeat > timeout) {
                node.status = .failed;
                try failed.append(node.id);
            }
        }
        
        return failed;
    }
};

test "一致性哈希路由" {
    const testing = std.testing;
    
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const allocator = gpa.allocator();
    
    var router = ConsistentHashRouter.init(allocator, 150);
    defer router.deinit();
    
    // 添加节点
    try router.addNode(.{
        .id = "node1",
        .address = "127.0.0.1",
        .port = 7001,
        .status = .active,
        .last_heartbeat = std.time.timestamp(),
    });
    
    try router.addNode(.{
        .id = "node2",
        .address = "127.0.0.1",
        .port = 7002,
        .status = .active,
        .last_heartbeat = std.time.timestamp(),
    });
    
    // 获取 Actor 所在节点
    const node = try router.getNode("user_001");
    try testing.expect(node.status == .active);
}

test "集群成员管理" {
    const testing = std.testing;
    
    var gpa = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = gpa.deinit();
    const allocator = gpa.allocator();
    
    const local_node = NodeInfo{
        .id = "node1",
        .address = "127.0.0.1",
        .port = 7001,
        .status = .active,
        .last_heartbeat = std.time.timestamp(),
    };
    
    var membership = ClusterMembership.init(allocator, local_node);
    defer membership.deinit();
    
    // 添加节点
    try membership.addNode(.{
        .id = "node2",
        .address = "127.0.0.1",
        .port = 7002,
        .status = .active,
        .last_heartbeat = std.time.timestamp(),
    });
    
    try testing.expect(membership.nodes.items.len == 1);
}
